//
// Created by martin on 2021/5/28.
//

#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include "fixedthreadpool.h"

/**
 * 线程池中每个线程对应的线程函数
 * @param arg [in] 线程参数
 * @note 线程函数的任务是, 当有空闲线程时, 就从等待执行的作业队列jobqueue取出用户作业job, 放入线程函数执行.
 * 每个线程同一时刻, 只能执行一个作业任务.
 */
static void *worker_callback(void *arg) {
    nWorker_t *worker= (nWorker_t *)arg;
    nFixedThreadPool *poolptr = worker->poolptr;

    while (1) {
        pthread_mutex_lock(&poolptr->mutex);

        /* 等待线程非空条件 */
        while (poolptr->jobqueue.front == NULL || poolptr->close == 0) {
            printf("thread %ld is waiting\n", pthread_self());
            pthread_cond_wait(&poolptr->notempty, &poolptr->mutex);
        }

        /* 线程池关闭时, 释放锁并退出当前线程 */
        if (poolptr->close == 1) {
            pthread_mutex_unlock(&poolptr->mutex);
            printf("thread %ld will exit\n", pthread_self());
            pthread_exit(NULL);
        }

        /* jobqueue弹出队头节点 */
        nJob_t *job = NULL;
        if (poolptr->jobqueue.front == poolptr->jobqueue.rear) {
            /* jobqueue只有一个节点 */
            job = poolptr->jobqueue.front;

            poolptr->jobqueue.front = poolptr->jobqueue.rear = NULL;
        }
        else {
            /* jobqueue 拥有超过1个节点 */
            job = poolptr->jobqueue.front;
            job->next = NULL;
            job->prev = NULL;

            poolptr->jobqueue.front = poolptr->jobqueue.front->next;
            poolptr->jobqueue.front->prev = poolptr->jobqueue.rear;
            poolptr->jobqueue.rear->next = poolptr->jobqueue.front;
        }

        pthread_mutex_unlock(&poolptr->mutex);
        pthread_cond_broadcast(&poolptr->notempty);

        /* 执行用户作业, 这里要求用户作业不能是无限循环, 不退出的作业, 否则线程无法正常结束, 从而导致线程池无法正常释放 */
        if (job)
            job->job_func(job->user_data);

        /* 释放用户作业节点 */
        if (job) {
            free(job);
            job = NULL;
        }
    }
}

/**
 * 创建指定线程数目的线程池
 * @param poolpptr [out] 指向线程池地址的指针
 * @param num 线程池线程数目
 * @return 创建结果
 * - < 0 创建失败
 * - 0 创建成功
 */
int fixedthreadpool_create(nFixedThreadPool **poolpptr, int num) {
    int i, ret;
    nFixedThreadPool *poolptr;

    if (poolpptr == NULL) return -1;
    /* 创建并验证线程池结构 */
    *poolpptr = (nFixedThreadPool *)malloc(sizeof (nFixedThreadPool));
    poolptr = *poolpptr;
    if (poolptr == NULL) {
        perror("malloc threadpool error");
        return -1;
    }

    /* 线程池线程数不能少于1个 */
    if (num < 1) num = 1;
    printf("fixedthreadpool_create, num = %d\n", num);

    /* 清0线程池结构 */
    memset(poolptr, 0, sizeof (nFixedThreadPool));

    /* 初始化 close */
    poolptr->close = 0;

    /* 初始化 mutex */
    pthread_mutex_init(&poolptr->mutex, NULL);

    /* 初始化条件变量 notempty */
    pthread_cond_init(&poolptr->notempty, NULL);

    /* 初始化工作队列 workerqueue : 按FIFO方式加入n个工作节点, 每个节点保存了线程标识符等信息 */
    for (i = 0; i < num; ++i) {
        printf("init %d-th worker of the pool\n", i);

        nWorker_t *worker = (nWorker_t *)malloc(sizeof (nWorker_t));
        if (worker == NULL) {
            perror("malloc error");
            return -2;
        }
        /* 清0工作节点 */
        memset(worker, 0, sizeof (nWorker_t));

        /* 当前工作节点绑定到当前线程池 */
        worker->poolptr = poolptr;

        /* 为每个工作节点创建1个线程, 并记录线程标识符信息 */
        ret = pthread_create(&worker->tid, NULL, worker_callback, (void *)worker);
        if (ret < 0) {
            perror("pthread_create error");
            free(worker); /* 释放本次创建的工作节点 */

            /* 释放已经存在的工作队列 */
            worker = poolptr->workerqueue.front;
            while (worker != poolptr->workerqueue.rear) {
                pthread_cancel(worker->tid); /* 尝试取消线程 */
                free(worker);
                worker = worker->next;
            }
            if (worker != NULL) {
                free(worker);
                worker = NULL;
            }

            /* 释放线程池结构 */
            free(poolptr);
            poolptr = NULL;

            return -3;
        }

        /* wokerqueue 末尾加入当前新建工作节点worker */
        if (poolptr->workerqueue.front == NULL || poolptr->workerqueue.rear == NULL ) {
            /* wokerqueue 是空队列 */
            worker->prev = worker;
            worker->next = worker;
            poolptr->workerqueue.front = worker;
        }
        else {
            /* wokerqueue 是非空队列 */
            worker->prev = poolptr->workerqueue.rear;
            worker->next = poolptr->workerqueue.front;

            poolptr->workerqueue.rear->next = worker;
            poolptr->workerqueue.front->prev = worker;
        }
        poolptr->workerqueue.rear = worker;
    }

    return 0;
}

/**
 * 向线程池加入一个用户作业
 * @param poolptr 线程池结构地址
 * @param callback 用户作业处理函数, 由用户自定义
 * @param arg 用户作业参数, 由用户定义
 * @return 加入用户作业结果
 * - < 0 加入失败
 * - 0 加入成功
 */
int fixedthreadpool_addjob(nFixedThreadPool *poolptr, jobcallback_t callback, void *arg) {
    if (poolptr == NULL) return -1;
    if (poolptr->workerqueue.front == NULL || poolptr->workerqueue.rear == NULL ) {
        fprintf(stderr, "workers is empty");
        return -1;
    }

    /* mutex 加锁 */
    pthread_mutex_lock(&poolptr->mutex);

    nJob_t *job = (nJob_t *)malloc(sizeof (nJob_t));
    if (job == NULL) {
        perror("malloc error");
        return -2;
    }
    job->next = NULL;
    job->prev = NULL;
    job->job_func = callback;
    job->user_data = arg;

    /* jobqueue 末尾加入当前作业节点job */
    if (poolptr->jobqueue.front == NULL || poolptr->jobqueue.rear == NULL) {
        /* jobqueue 是空队列*/
        job->next = job;
        job->prev = job;

        poolptr->jobqueue.front = job;
    }
    else {
        /* jobqueue 是非空队列*/
        job->prev = poolptr->jobqueue.rear;
        job->next = poolptr->jobqueue.front;

        poolptr->jobqueue.rear->next = job;
        poolptr->jobqueue.front->prev = job;
    }
    poolptr->jobqueue.rear = job;

    /* mutex 解锁, 并激活等待notempty的所有线程 */
    pthread_mutex_unlock(&poolptr->mutex);
    pthread_cond_broadcast(&poolptr->notempty);

    return 0;
}

/**
 * 销毁线程池
 * @param poolptr 线程池结构地址
 * @note 销毁线程池结构所有指针指向的区域, 包括工作队列, 作业队列, 互斥量, 条件变量等. 正在执行作业任务的线程先待其执行完毕,
 * 未执行的作业任务直接释放其资源
 */
void fixedthreadpool_destroy(nFixedThreadPool *poolptr) {
    if (poolptr == NULL) return;

    /* 置线程池关闭状态 */
    poolptr->close = 1;
    // wakeup all blocked thread of the pool
    /* 唤醒所有等待条件变量notempty的线程 */
    pthread_cond_broadcast(&poolptr->notempty);

    /* 连接工作队列线程, 线程执行完毕后释放工作队列节点 */
    nWorker_t *worker = poolptr->workerqueue.front;
    while (worker != NULL && worker != poolptr->workerqueue.rear) {
        pthread_join(worker->tid, NULL);

        free(worker);
        worker = worker->next;
    }
    if (worker != NULL) {
        /* 队列尾节点 */
        free(worker);
        worker = NULL;
    }

    /* 直接释放待执行的作业队列, 而不继续执行 */
    nJob_t *job = poolptr->jobqueue.front;
    while (job != NULL && job != poolptr->jobqueue.rear) {
        free(job);
        job = job->next;
    }
    if (job != NULL) {
        /* 队列尾节点 */
        free(job);
        job = NULL;
    }

    /* 销毁互斥量, 条件变量 */
    pthread_cond_destroy(&poolptr->notempty);
    pthread_mutex_destroy(&poolptr->mutex);

    /* 销毁线程池结构*/
    free(poolptr);
    poolptr = NULL;
}
